package com.warrior.superdata.dds.esclient;

import com.google.common.collect.Maps;
import com.warrior.superdata.dds.domain.DBVo;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.util.StringUtils;

import java.util.Iterator;
import java.util.Map;

/**
 * @Author xujian
 * @Description 动态es连接，里面可以放不同host的es连接
 * @Date 9:23 2020/8/27
 **/
@Slf4j
public class ElasticsearchRestClientHolder {

    private static volatile ElasticsearchRestClientHolder poll = null;

    private static Map<String, RestClientTimer> ES_CLIENT = Maps.newHashMap();

    /**
     * @Author xujian
     * @Description 双重校验锁的单例模式
     * @Date 9:24 2020/8/27
     * @return com.warrior.businesscenter.common.esclient.ElasticsearchRestClientPool
     **/
    public static ElasticsearchRestClientHolder getInstance() {
        if (null == poll) {
            synchronized (ElasticsearchRestClientHolder.class) {
                if (null == poll) {
                    poll = new ElasticsearchRestClientHolder();
                }
            }
        }
        return poll;
    }

    /**
     * @Author xujian
     * @Description 根据每个host+port获取每个es的连接
     * @Date 9:25 2020/8/27
     * @Param [dbDto]
     * @return com.warrior.businesscenter.common.esclient.RestClientTimer
     **/
    public RestClientTimer getRestClientTimer(DBVo dbDto, Boolean isCache) {
        try {
            if (StringUtils.isEmpty(dbDto.getIp()) || StringUtils.isEmpty(dbDto.getPort())) {
                return null;
            }
            RestClientTimer restClient = ES_CLIENT.get(dbDto.getUrl());
            if (null == restClient || !isCache) {
                ElasticsearchRestClient client = new ElasticsearchRestClient();
                client.restClientBuilder(dbDto);
                RestClientTimer clientTimer = new RestClientTimer(client);
                ES_CLIENT.put(dbDto.getUrl(), clientTimer);
                return clientTimer;
            }
            restClient.refreshTime();
            return restClient;
        } catch (Exception e) {
            log.error("=========== 获取RestClienBuilder异常 errMsg:{} ==========", e.getMessage());
        } finally {
            checkDelRestClient();
        }
        return null;
    }

    /**
     * @Author xujian
     * @Description 获取ElasticsearchRestClient
     * @Date 9:26 2020/8/27
     * @Param [dbDto]
     * @return com.warrior.businesscenter.common.esclient.ElasticsearchRestClient
     **/
    public ElasticsearchRestClient getElasticsearchRestClient(DBVo dbDto, Boolean isCache) {
        RestClientTimer restClientTimer = getRestClientTimer(dbDto, isCache);
        if (null != restClientTimer) {
            return restClientTimer.getRestClient();
        }
        return null;
    }

    /**
     * @Author xujian
     * @Description 获取RestClientBuilder
     * @Date 9:26 2020/8/27
     * @Param [dbDto]
     * @return org.elasticsearch.client.RestClientBuilder
     **/
    public RestClientBuilder getRestClientBuilder(DBVo dbDto, Boolean isCache) {
        ElasticsearchRestClient restClient = getElasticsearchRestClient(dbDto, isCache);
        if (null != restClient) {
            return restClient.getRestClientBuilder();
        }
        return null;
    }

    /**
     * @Author xujian
     * @Description 获取RestHighLevelClient
     * @Date 9:26 2020/8/27
     * @Param [dbDto]
     * @return org.elasticsearch.client.RestHighLevelClient
     **/
    public RestHighLevelClient getRestHighLevelClient(DBVo dbDto, Boolean isCache) {
        ElasticsearchRestClient restClient = getElasticsearchRestClient(dbDto, isCache);
        if (null != restClient) {
            return restClient.getRestHighLevelClient();
        }
        return null;
    }

    /**
     * @Author xujian
     * @Description 释放过期的连接
     * @Date 9:27 2020/8/27
     * @Param []
     * @return void
     **/
    private synchronized void checkDelRestClient() {
        if (ES_CLIENT.size() != 0) {
            Iterator<Map.Entry<String, RestClientTimer>> it = ES_CLIENT.entrySet().iterator();
            while (it.hasNext()) {
                Map.Entry<String, RestClientTimer> entry = it.next();
                if (entry.getValue().checkAndClose()) {
                    it.remove();
                }
            }
        }
    }
}
